1. FastAPI 是什么?
FastAPI 是一个用于构建 API 的现代、高性能的 Python Web 框架。它的名字已经点明了其两大核心特点:
- Fast (快速运行):基于 Starlette (ASGI 框架) 和 Pydantic (数据校验库),性能与 NodeJS 和 Go 不相上下,是 Python 主流框架中性能最高的之一。
- Fast (快速开发):得益于 Python 的类型提示(Type Hints),FastAPI 极大地提升了开发效率、减少了 Bug,并且能提供强大的编辑器支持(如自动补全)。
它不仅仅是一个框架,更是一套完整的 API 开发解决方案。
2. 为什么选择 FastAPI?(核心优势)
- 自动交互式文档:这是 FastAPI 的杀手级功能。你只需要编写正常的 Python 代码,它就能自动生成两种交互式 API 文档界面(Swagger UI 和 ReDoc)。你可以在浏览器中直接测试你的 API,无需任何额外配置。
- 基于标准:完全兼容并基于 OpenAPI (前身为 Swagger) 和 JSON Schema 这两个开放标准。这意味着你的 API 可以轻松地与各种工具集成。
- 类型安全与数据校验:利用 Pydantic,你可以用纯 Python 类型来定义数据模型。FastAPI 会自动完成:
- 数据解析:将请求中的 JSON 数据转换成 Python 对象。
- 数据校验:如果请求数据不符合你定义的类型(例如,本应是
int
却收到了string
),FastAPI 会自动返回一个清晰的 422 错误。 - 数据序列化:将你返回的 Python 对象自动转换成 JSON 响应。
- 强大的依赖注入系统:这是一个非常优雅和强大的功能,可以轻松处理认证、数据库会话管理、权限控制等复杂逻辑,让代码更模块化、更易于测试。
- 异步支持:原生支持
async/await
语法,非常适合处理 I/O 密集型任务(如数据库查询、调用外部 API),从而获得极高的并发性能。同时,它也兼容普通的同步函数。 - 出色的编辑器支持:由于大量使用类型提示,VS Code、PyCharm 等现代编辑器能为你提供无与伦比的自动补全和类型检查体验,编码时如虎添翼。
3. 入门实践:第一个 FastAPI 应用
第 1 步:安装
你需要安装 FastAPI 和一个 ASGI 服务器,最常用的是 Uvicorn。
pip install fastapi uvicorn[standard]
[standard]
会额外安装一些推荐的依赖,如uvloop
和httptools
,能进一步提升性能。
第 2 步:编写代码 (main.py
)
创建一个名为 main.py
的文件,并写入以下代码:
from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional
# 1. 创建 FastAPI 实例
app = FastAPI()
# 2. 定义数据模型 (使用 Pydantic)
class Item(BaseModel):
name: str
description: Optional[str] = None # 可选字段,默认值为 None
price: float
tax: Optional[float] = None
# 3. 创建路径操作 (也叫路由或端点)
# @app 是一个装饰器,它将下面的函数注册为一个 API 端点
# .get("/") 表示这个端点响应 HTTP GET 请求,路径是 "/"
@app.get("/")
def read_root():
return {"message": "Hello, FastAPI World!"}
# 带路径参数的 GET 请求
@app.get("/items/{item_id}")
def read_item(item_id: int, q: Optional[str] = None):
# item_id 会被自动从字符串转换为整数,并进行校验
# q 是一个可选的查询参数,例如 /items/5?q=somequery
return {"item_id": item_id, "q": q}
# 接收请求体的 POST 请求
@app.post("/items/")
def create_item(item: Item):
# FastAPI 会自动将请求的 JSON body 解析、校验并填充到 item 对象中
item_dict = item.dict()
if item.tax:
price_with_tax = item.price + item.tax
item_dict.update({"price_with_tax": price_with_tax})
return item_dict
第 3 步:运行服务器
在终端中,进入 main.py
所在的目录,运行以下命令:
uvicorn main:app --reload
main
: 指的是main.py
文件。app
: 指的是你在main.py
中创建的FastAPI()
实例。--reload
: 这个参数会让服务器在代码变动后自动重启,非常适合开发阶段。
现在,你的 API 已经在本地运行了!
第 4 步:体验交互式文档
打开浏览器,访问以下两个 URL:
- Swagger UI:
http://127.0.0.1:8000/docs
- ReDoc:
http://127.0.0.1:8000/redoc
你会看到一个完整的 API 文档界面。在这里,你可以看到所有的 API 端点、它们的参数、请求体格式、响应格式等。你甚至可以点击 “Try it out”,直接在页面上发送请求并看到结果!
4. 核心功能详解
4.1 路径参数 (Path Parameters)
在路径中使用花括号 {}
来声明,并作为函数的参数接收。FastAPI 会利用类型提示进行转换和校验。
@app.get("/users/{user_id}")
def get_user(user_id: int):
return {"user_id": user_id}
如果访问 /users/abc
,你会自动收到一个 422 错误,因为 abc
无法被转换成 int
。
4.2 查询参数 (Query Parameters)
函数参数中,没有出现在路径里的,会自动被识别为查询参数。
@app.get("/search/")
# skip 是查询参数,有默认值0;limit 也是,有默认值100
def search_items(skip: int = 0, limit: int = 100):
return {"skip": skip, "limit": limit}
访问 http://127.0.0.1:8000/search/?skip=10&limit=50
即可。
4.3 请求体 (Request Body)
使用 Pydantic 的 BaseModel
定义数据结构,并将其作为函数参数。FastAPI 会自动处理 JSON 的解析和校验。
from pydantic import BaseModel
class User(BaseModel):
username: str
email: str
is_active: bool = True
@app.post("/users/")
def create_user(user: User):
# 此时 user 是一个 User 类的实例,可以直接访问属性
print(user.username)
return user
当你发送一个 POST 请求到 /users/
时,请求体需要是类似 {"username": "john.doe", "email": "johndoe@example.com"}
的 JSON。
4.4 混合使用
你可以将路径参数、查询参数和请求体混合在一个函数中使用。
@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item, q: Optional[str] = None):
# item_id 来自路径
# item 来自请求体
# q 来自查询参数
result = {"item_id": item_id, **item.dict()}
if q:
result.update({"q": q})
return result
5. 高级特性
5.1 依赖注入 (Dependency Injection)
这是 FastAPI 最强大的特性之一。通过 Depends
,你可以将一些共享的逻辑(如数据库连接、用户认证)注入到你的路径操作函数中。
示例:一个简单的共享参数依赖
from fastapi import Depends, FastAPI
app = FastAPI()
# 这是一个依赖项 (一个可调用的函数)
async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
# 使用 Depends 将 common_parameters 的返回值注入到 commons 变量中
async def read_items(commons: dict = Depends(common_parameters)):
return {"message": "Items data", "params": commons}
@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
return {"message": "Users data", "params": commons}
在这个例子中,common_parameters
定义了一组通用的查询参数。任何需要这组参数的端点都可以通过 Depends
来获取它们,避免了代码重复。
更实际的例子:数据库连接
# 伪代码
def get_db_session():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db_session)):
# db 就是一个数据库会话,函数执行完后会自动关闭
user = db.query(User).filter(User.id == user_id).first()
return user
5.2 安全与认证
FastAPI 内置了处理安全问题的工具,例如 OAuth2、API Keys 等。
示例:使用 OAuth2 进行密码流验证
from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer
app = FastAPI()
# 定义安全方案:它会查找请求头中的 "Authorization: Bearer <token>"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
@app.get("/users/me")
# FastAPI 会自动验证 Header,并将 token 注入到函数中
async def read_users_me(token: str = Depends(oauth2_scheme)):
return {"token": token}
这只是基础,FastAPI 官方文档有完整的教程来教你如何实现一个完整的登录和令牌验证系统。
5.3 响应模型 (Response Model)
你可以指定一个模型用于响应,这有两个好处:
- 数据过滤:只返回模型中定义的字段,即使你的函数返回了更多的数据(比如密码哈希)。
- 文档生成:API 文档会精确地显示响应的数据结构。
class UserIn(BaseModel): # 用于输入
username: str
password: str
class UserOut(BaseModel): # 用于输出
username: str
email: str
@app.post("/users/", response_model=UserOut)
def create_user(user: UserIn):
# 假设这里创建了用户并存入数据库
# ...
# 返回的数据包含密码,但 FastAPI 会根据 response_model=UserOut 进行过滤
return {"username": user.username, "email": "test@example.com", "password": "hashed_password"}
客户端只会收到 {"username": "...", "email": "..."}
,密码被自动过滤掉了。
6. 项目结构化
当应用变大时,把所有代码放在一个 main.py
文件里是不现实的。FastAPI 提供了 APIRouter
来帮助你组织代码。
建议的目录结构:
.
├── app/
│ ├── __init__.py
│ ├── main.py # 主应用文件,组装所有路由
│ ├── crud.py # 数据库操作函数 (Create, Read, Update, Delete)
│ ├── database.py # 数据库连接配置
│ ├── models.py # SQLAlchemy 的数据库模型
│ ├── schemas.py # Pydantic 的数据模型 (请求体、响应体)
│ └── routers/
│ ├── __init__.py
│ ├── users.py # 处理用户相关的路由
│ └── items.py # 处理物品相关的路由
app/routers/users.py
示例:
from fastapi import APIRouter, Depends
from .. import schemas # 使用相对导入
# 创建一个 APIRouter 实例
router = APIRouter(
prefix="/users", # 为这个路由下的所有路径添加前缀 /users
tags=["users"], # 在 API 文档中为这组路由添加标签
)
@router.get("/", response_model=list[schemas.User])
def read_users():
# ... 逻辑 ...
return [{"username": "user1"}, {"username": "user2"}]
app/main.py
中组装路由:
from fastapi import FastAPI
from .routers import users, items
app = FastAPI()
# 将 users 路由和 items 路由包含进来
app.include_router(users.router)
app.include_router(items.router)
@app.get("/")
def root():
return {"message": "Welcome to the main application"}